package sfs

import (
	"context"
	"crypto/sha1"
	"errors"
	"fmt"
	"github.com/chicken-team-outside/chicken_transmission/api_messages"
	"github.com/chicken-team-outside/chicken_transmission/socket"
	"github.com/golang/protobuf/proto"
	"go.uber.org/zap"
	"io"
	"io/fs"
	"strings"
	"sync"
	"sync/atomic"
	"time"
)

type OperatorKeeperClient struct {
	ctx        context.Context
	cancel     context.CancelCauseFunc
	client     *FsClient
	replyCtx   *socket.ReplayContext
	timeout    int64
	lastPing   atomic.Int64
	resultChan chan []byte
	once       sync.Once
}

func (h *OperatorKeeperClient) sendPong(clientId, replyId uint64) error {
	_, err := h.client.send(h.ctx, clientId, replyId, api_messages.MessageType_OperatorPong, nil, false)
	if err != nil {
		return err
	}
	return nil
}

func (h *OperatorKeeperClient) handler(message *socket.Message) {
	switch message.Header.Type {
	case api_messages.MessageType_OperatorErr:
		errMsg := new(api_messages.OperatorErrMessage)
		err := proto.Unmarshal(message.Body, errMsg)
		if err != nil {
			zap.L().Error("receive [OperatorErr] CMD", zap.Error(err), zap.Any("message", message))
			h.cancel(err)
		} else {
			h.cancel(errorToError(errMsg.ErrMsg))
		}
	case api_messages.MessageType_OperatorSuccess:
		zap.L().Debug("receive [OperatorSuccess] CMD", zap.Any("message", message))
		select {
		case h.resultChan <- message.Body:
		case <-h.ctx.Done():
		}
		return
	case api_messages.MessageType_OperatorPing:
		reply := new(api_messages.PingMessage)
		err := proto.Unmarshal(message.Body, reply)
		zap.L().Debug("receive [OperatorPing] CMD", zap.Any("message", message), zap.Error(err))
		if reply.ReplyId > 0 {
			h.lastPing.Store(time.Now().Unix())
			err = h.sendPong(message.Header.ClientId, reply.ReplyId)
			zap.L().Debug("send [OperatorPong] CMD", zap.Any("header", message.Header), zap.Uint64("reply", reply.ReplyId), zap.Error(err))
			if err != nil && !errors.Is(err, context.Canceled) {
				zap.L().Warn("failed to send [OperatorPong] CMD", zap.Any("message", message), zap.Error(err))
			}
		}
	default:
		zap.L().Debug("receive [Unknown CMD]", zap.Any("message", message))
	}
}

func (h *OperatorKeeperClient) QueryAndWait(clientId uint64, t api_messages.MessageType, content proto.Message) ([]byte, error) {
	var err error
	h.replyCtx, err = h.client.rpcClient.SendProtobuf(h.ctx, h.timeout, clientId, 0, t, content, true)
	if err != nil {
		return nil, err
	}
	defer h.Close()
	go func() {
		tk := time.NewTicker(time.Duration(h.timeout) * time.Second)
		defer tk.Stop()
		h.lastPing.Store(time.Now().Unix())
		for {
			select {
			case <-tk.C:
			case <-h.ctx.Done():
				return
			case msg := <-h.replyCtx.ReplyChan:
				if msg != nil {
					h.handler(msg)
				} else {
					return
				}
			}
			if h.lastPing.Load() < time.Now().Unix()-h.timeout {
				h.cancel(socket.ErrTimeout)
				return
			}
		}
	}()
	select {
	case result := <-h.resultChan:
		return result, nil
	case <-h.ctx.Done():
		err := context.Cause(h.ctx)
		if err == nil {
			err = h.ctx.Err()
		}
		return nil, err
	}
}

func (h *OperatorKeeperClient) Close() {
	h.once.Do(func() {
		h.cancel(context.Canceled)
		if h.replyCtx != nil {
			h.replyCtx.Close()
		}
	})
}

func NewOperatorKeeperClient(ctx context.Context, client *FsClient, timeout int64) *OperatorKeeperClient {
	fh := &OperatorKeeperClient{
		client:     client,
		timeout:    timeout,
		resultChan: make(chan []byte, 1),
	}
	fh.ctx, fh.cancel = context.WithCancelCause(ctx)
	return fh
}

type OperatorKeeperServer struct {
	ctx         context.Context
	cancel      context.CancelCauseFunc
	clientId    uint64
	replyId     uint64
	timeout     int32
	client      *FsClient
	lastPong    atomic.Int64
	pingReplyId uint64
}

func (h *OperatorKeeperServer) sendPing(first bool) error {
	msgId := h.client.rpcClient.GetMsgId()
	if first {
		h.pingReplyId = msgId
	}
	msgCtx, err := h.client.rpcClient.SendProtobufCustom(h.ctx, int64(h.timeout), &api_messages.MessageHeader{
		Id:       msgId,
		ClientId: h.clientId,
		Reply:    h.replyId,
	}, &api_messages.PingMessage{ReplyId: h.replyId}, first)
	if err != nil {
		return err
	}
	if first {
		h.lastPong.Store(time.Now().Unix())
	}
	if first {
		go func() {
			defer msgCtx.Close()
			for {
				select {
				case <-h.ctx.Done():
					return
				case message := <-msgCtx.ReplyChan:
					if message.Header.Type == api_messages.MessageType_OperatorPong {
						zap.L().Debug("received  [OperatorPong] CMD", zap.Any("message", message))
						h.lastPong.Store(time.Now().Unix())
					}
				}
			}
		}()
	}
	return nil
}

func (h *OperatorKeeperServer) keepAlive() {
	err := h.sendPing(true)
	if err != nil {
		h.cancel(fmt.Errorf("failed to send first ping: %w", err))
		return
	}
	tik := time.NewTimer(time.Duration(max(h.timeout>>1, 1)) * time.Second)
	defer tik.Stop()
	for {
		select {
		case <-tik.C:
			if time.Now().Unix()-h.lastPong.Load() >= int64(h.timeout) {
				h.cancel(fmt.Errorf("last ping timeout: %w", socket.ErrTimeout))
				return
			} else {
				err = h.sendPing(false)
				if err != nil {
					h.cancel(fmt.Errorf("failed to send ping: %w", err))
					return
				}
			}
		case <-h.ctx.Done():
			return
		}
	}
}

func (h *OperatorKeeperServer) sendError(err error) error {
	_, err = h.client.send(h.ctx, h.clientId, h.replyId, api_messages.MessageType_OperatorErr, &api_messages.OperatorErrMessage{
		ErrMsg: err.Error(),
	}, false)
	if err != nil {
		return err
	}
	return nil
}

func (h *OperatorKeeperServer) success(data proto.Message) error {
	_, err := h.client.rpcClient.SendProtobuf(h.ctx, int64(h.timeout), h.clientId, h.replyId, api_messages.MessageType_OperatorSuccess, data, false)
	if err != nil {
		return err
	}
	return nil
}

func (h *OperatorKeeperServer) Close() {
	h.cancel(context.Canceled)
}

func NewOperatorKeeperServer(ctx context.Context, clientId, replay uint64, client *FsClient, timeout int32) *OperatorKeeperServer {
	fh := &OperatorKeeperServer{
		clientId: clientId,
		replyId:  replay,
		client:   client,
		timeout:  timeout,
	}
	fh.ctx, fh.cancel = context.WithCancelCause(ctx)
	return fh
}

type atomicValue struct {
	Value interface{}
}

func errorToError(desc string) error {
	if strings.HasSuffix(desc, ": The system cannot find the file specified.") {
		return fmt.Errorf("%s: %w", desc, fs.ErrNotExist)
	}
	return errors.New(desc)
}

func decodeStatusMessageDirect(s *api_messages.StatusMessage) error {
	if s.Success {
		return nil
	} else {
		return errorToError(s.Message)
	}
}

func decodeStatusMessage(r []byte) error {
	status := new(api_messages.StatusMessage)
	err := proto.Unmarshal(r, status)
	if err == nil {
		return decodeStatusMessageDirect(status)
	}
	return err
}

func encodeStatusMessage(err error) (r *api_messages.StatusMessage) {
	r = new(api_messages.StatusMessage)
	if err != nil {
		r.Success = false
		r.Message = err.Error()
	} else {
		r.Success = true
	}
	return
}

func CalculateFileBlocksSha1(reader io.Reader, size int64, blockSize int32) (sha1List []byte, err error) {
	buf := make([]byte, 64*1024)
	m := sha1.New()
	var readNum int
	blockNum := size / int64(blockSize)
	if size%int64(blockSize) > 0 {
		blockNum += 1
	}
	sha1List = make([]byte, 0, blockNum*sha1.Size)
	remaining := blockSize
	for err == nil {
		readNum, err = reader.Read(buf)
		remaining -= int32(readNum)
		if remaining <= 0 {
			end := int32(readNum) + remaining
			m.Write(buf[:end])
			sha1List = m.Sum(sha1List)
			m = sha1.New()
			m.Write(buf[end:])
			remaining = blockSize + remaining
		} else {
			m.Write(buf[:readNum])
		}
		if err == io.EOF {
			if remaining != blockSize {
				sha1List = m.Sum(sha1List)
			}
			err = nil
			break
		}
	}
	return
}

func ReadHttpErrorMsg(reader io.ReadCloser) string {
	if reader != nil {
		defer reader.Close()
		buf := make([]byte, 1024)
		n, _ := reader.Read(buf)
		return string(buf[:n])
	}
	return ""
}
